package cn.tedu.micro.demo.client;

import com.rabbitmq.client.*;
import lombok.extern.slf4j.Slf4j;
import org.junit.jupiter.api.Test;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeoutException;

/**
 * 把入门中 发送 消费 声明的代码全部都在这一个类中测试方法实现
 */
@Slf4j
public class RabbitmqTutorail {
    //测试方法 声明组件 声明一个叫做queue01的队列
    @Test
    public void declareQueue01() throws IOException, InterruptedException {
        //1.获取操作rabbitmq的短连接
        Channel channel = RabbitChannelManager.newChannel();
        //2.声明queue01
        /**
         *String queue: 创建的队列名称,只能存在唯一名称
         *Boolean durable: 队列是否持久化 true在rabbitmq重启之后 队列保持存在 false不存在
         *Boolean exclusive: 队列是绑定当前连接对象(独占 专属) true别的链接无法操作队列 false所有连接都可以操作队列
         *Boolean autoDelete: 队列是否自动删除 true队列从第一个监听消费者开始计算,到最后一个消费者断开连接,自动删除 false不删除
         *Map<String,Object> args: key值和value值 可以说是固定的,能描述队列其他特性
         *描述声明队列的最大长度:   x-max_length
         *队列存活时间 x-expires: 如果没人用队列,最多可以存在多久毫秒数
         */
        channel.queueDeclare("queue01", false, false, false, null);
        /*Map<String,Object> args=new HashMap<>();
        args.put("x-max-length",10);
        args.put("x-expires",10000);
        channel.queueDeclare(
                "queue01",
                false,
                false,
                false,
                args);
        Thread.sleep(10000000);*/
    }
    //向queue01里发送一条消息
    @Test
    public void sendMsg2Queue01() throws IOException, TimeoutException {
        //1.获取连接
        Channel channel = RabbitChannelManager.newChannel();
        //2.组织消息
        //2.1 消息体
        byte[] body = "hello rabbitmq".getBytes(StandardCharsets.UTF_8);
        //2.2 消息属性 写一个字符集编码 数据类型 携带一个自定义属性name=王翠花
        AMQP.BasicProperties.Builder builder=new AMQP.BasicProperties.Builder();
        //固定属性
        builder.contentEncoding("UTF-8");
        builder.contentType("text/plain");
        //自定义属性
        Map<String,Object> headers=new HashMap<>();
        headers.put("name","李守富");
        headers.put("age",18);
        builder.headers(headers);
        //使用builder获取最终属性对象
        AMQP.BasicProperties properties = builder.build();
        //2.3消息路由key 想要发送消息到queue01
        String routingKey="queue01";
        //3.调用api发送消息
        //3.1任何生产者都要将消息发送到交换机 第一个参数是一个交换机名称exchange
        channel.basicPublish("",routingKey,properties,body);
        //4.使用完毕 关闭连接
        channel.close();
    }
    //从queue01监听消息,消费打印
    @Test
    public void consumerQueue01() throws IOException, InterruptedException {
        Channel channel = RabbitChannelManager.newChannel();
        //1.准备一个消费回调,每当有一条消息传递给这个channel是回调方法处理的消费逻辑
        DeliverCallback deliver=new DeliverCallback() {
            /**
             * 方法作用: 只要监听队列 有消息,就会给当前这个消费者推送消息
             * 每条消息,在消费者中handle处理
             * String consumerTag: 消费者身份 rabbitmq分配的表示tag
             * Delivery message: 就是消息对象 发了什么 这里就能收到什么
             */
            @Override
            public void handle(String consumerTag, Delivery message) throws IOException {
                //1.消息体byte
                byte[] body = message.getBody();
                String msg=new String(body, StandardCharsets.UTF_8);
                log.info("msg:"+msg);
                //2.消息属性
                AMQP.BasicProperties properties = message.getProperties();
                log.info("properties.getContentEncoding() = " + properties.getContentEncoding());
                log.info("properties.getContentType() = " + properties.getContentType());
                log.info("properties.getHeaders() = " + properties.getHeaders());
                //3.消息路由
                String routingKey = message.getEnvelope().getRoutingKey();
                log.info("routingKey = " + routingKey);
            }
        };
        CancelCallback cancel=new CancelCallback() {
            /**
             * 消费者取消消费的回调
             * String consumerTag: 消费者标签
             * cancel一般是释放资源用的
             */
            @Override
            public void handle(String consumerTag) throws IOException {
                log.info("consumerTag = " + consumerTag);
            }
        };
        //3.调用监听队列的方法
        //queue: 消费者想要监听的队列 queue01
        //autoAck: 和消费确认有关,目前是自动确认
        //deliver: 消息入口处理对象
        //cancel: 消费者取消消费的回调
        channel.basicConsume("queue01",true,deliver,cancel);
        //4看到消费过程,需要当前测试线程持续运行
        Thread.sleep(10000000);
    }


}
